package com.shujia.dwd

import java.awt.geom.Point2D
import java.text.SimpleDateFormat

import com.shujia.grid.Grid
import com.shujia.util.SparkTool
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.mutable.ListBuffer

object DwdStayPointApp extends SparkTool {
  /**
    * 在子类中实现run方法，实现自定义的代码累哦及
    *
    * 在子类run方法的前面加上下面两行代码
    * import spark.implicits._
    * import org.apache.spark.sql.functions._
    *
    * @param spark spark的环境
    */
  override def run(spark: SparkSession): Unit = {

    import spark.implicits._

    //1、读取融合表的数据,取出所需要的字段
    val dataDF: DataFrame = spark.sql(
      s"""
         |select
         |mdn, -- 手机号
         |start_time, -- 业务时间，20180503211049,20180503210349
         |county_id, -- 停留点的区县编码
         |grid_id -- 网格编号
         |from dwd.dwd_res_regn_mergelocation_msk_d  where day_id=$day_id
         |
         |""".stripMargin)

    //1、按照手机号进行分组

    val kvRDD: RDD[(String, (String, String, String, String))] = dataDF
      .rdd
      .map {
        case Row(mdn: String, start_time: String, county_id: String, grid_id: String) =>

          val split: Array[String] = start_time.split(",")
          val startTime: String = split(1)
          val endTime: String = split(0)

          //以手机号作为key
          (mdn, (grid_id, startTime, endTime, county_id))
      }

    val groupByKeyRDD: RDD[(String, Iterable[(String, String, String, String)])] = kvRDD.groupByKey()


    /**
      * 按照时间循序对数据进行聚类
      *
      */
    val juleiRDD: RDD[(String, String, String, String, String)] = groupByKeyRDD.flatMap {
      case (mdn: String, points: Iterable[(String, String, String, String)]) =>

        //一个人一天的所有的数据
        val pointList: List[(String, String, String, String)] = points.toList


        //按照时间进行升序排序
        val sortPoint: List[(String, String, String, String)] = pointList.sortBy(_._2)


        /**
          * 取出第一条数据
          *
          */

        val lastPoint: (String, String, String, String) = sortPoint.head
        var lastGrid: String = lastPoint._1
        var lastStartTime: String = lastPoint._2
        var lastEndTime: String = lastPoint._3
        var lastCountyId: String = lastPoint._4

        //存放最终的结果
        val resultPoint = new ListBuffer[(String, String, String, String, String)]

        //tail 取出不包括第一条数据的所有的数据
        sortPoint.tail.foreach {
          case (grid_id: String, startTime: String, endTime: String, county_id: String) =>

            if (lastGrid.equals(grid_id)) {

              //如果第一条和第二条是同一个网格，
              lastEndTime = endTime
            } else {

              //如果网格不一样了，保证前面的数据
              resultPoint += ((mdn, lastGrid, lastStartTime, lastEndTime, lastCountyId))

              //切换网格
              lastGrid = grid_id
              lastStartTime = startTime
              lastEndTime = endTime
              lastCountyId = county_id
            }
        }

        //处理最后一个组
        resultPoint += ((mdn, lastGrid, lastStartTime, lastEndTime, lastCountyId))

        //返回结果
        resultPoint.toList

    }

    /**
      * mdn string comment '用户手机号码'
      * ,longi string comment '网格中心点经度'
      * ,lati string comment '网格中心点纬度'
      * ,grid_id string comment '停留点所在电信内部网格号'
      * ,county_id string comment '停留点区县'
      * ,duration string comment '机主在停留点停留的时间长度（分钟）,lTime-eTime'
      * ,grid_first_time string comment '网格第一个记录位置点时间（秒级）'
      * ,grid_last_time string comment '网格最后一个记录位置点时间（秒级）'
      */

    //整理数据
    val stayPointRDD: RDD[(String, String, String, String, String, String, String, String)] = juleiRDD.map {
      case (mdn: String, grid: String, startTime: String, endTime: String, countyId: String) =>

        //获取网格中心点的经纬度
        val centerPoint: Point2D.Double = Grid.getCenter(grid.toLong)

        val longi: String = centerPoint.getX.formatted("%.4f")
        val lati: String = centerPoint.getY.formatted("%.4f")

        //计算停留时间
        val format = new SimpleDateFormat("yyyyMMddHHmmss")
        val duration: Double = (format.parse(endTime).getTime - format.parse(startTime).getTime) / 60000.0

        (mdn, longi, lati, grid, countyId, duration.formatted("%.1f"), startTime, endTime)
    }



    //保存数据
    saveDataAndAddPartitionWithDay(
      stayPointRDD.toDF(),
      "/daas/motl/dwd/dwd_staypoint_msk_d",
      "dwd.dwd_staypoint_msk_d"
    )

  }
}
